package com.bitters.kafka;

import com.alibaba.fastjson.JSON;
import com.bitters.kafka.config.KafkaConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.HashMap;
import java.util.Map;

/**
 * Created with IntelliJ IDEA
 * Created By thomas
 * Date: 2019/5/9
 * Description ${description}
 * Time: 16:10
 */
@Component
public class KafkaSenderMain<T> {

    private Logger logger = LoggerFactory.getLogger(KafkaSenderMain.class);


    @Autowired
    private KafkaConfig kafkaConfig;

    private Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap();
        props.put("bootstrap.servers", this.kafkaConfig.servers);
        props.put("retries", 3);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", StringSerializer.class);
        props.put("value.serializer", StringSerializer.class);
        return props;
    }

    private ProducerFactory<String, Object> producerFactory() {
        return new DefaultKafkaProducerFactory(this.producerConfigs());
    }

    @Bean
    @Primary
    public KafkaTemplate<String, Object> kafkaTemplate() {
        KafkaTemplate template = new KafkaTemplate(this.producerFactory());
        return template;
    }

    public void send(T object) {
        String jsonObj = JSON.toJSONString(object);
        this.logger.info("------------ message = {}", jsonObj);
        ListenableFuture<SendResult<String, Object>> future = this.kafkaTemplate().send("kafka1", jsonObj);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            public void onFailure(Throwable throwable) {
                KafkaSenderMain.this.logger.info("Produce: The message failed to be sent:" + throwable.getMessage());
            }

            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
                KafkaSenderMain.this.logger.info("Produce: The message was sent successfully:");
                KafkaSenderMain.this.logger.info("Produce: @@@@@@^_^^_^^_^@@@@@@result: " + stringObjectSendResult.toString());
            }
        });
    }
}
